package streamAPI.asyncio;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.testng.annotations.Test;

import java.util.concurrent.TimeUnit;

/**
 * 通过异步IO 查询
 *
 * https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/stream/operators/asyncio.html
 * http://hc.apache.org/httpcomponents-asyncclient-dev/quickstart.html
 * @author yue.cao
 * @since 10-21-2020
 */
public class AsyncIOProject {

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.socketTextStream("caoyuesh1", 4444);
		SingleOutputStreamOperator<Tuple2<String, String>> res =
				//最多10个  超过10个就阻塞
				AsyncDataStream.unorderedWait(source, new AsyncGetWebRequest(), 0, TimeUnit.MICROSECONDS, 10);

		res.print();
		env.execute();
	}

	@Test(description = "from mysql")
	public void t1() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> source = env.socketTextStream("caoyuesh1", 4444);
		SingleOutputStreamOperator<Tuple2<String, String>> res =
				//最多10个  超过10个就阻塞
				AsyncDataStream.unorderedWait(source, new AsyncMySQLRequet(), 0, TimeUnit.MICROSECONDS, 1);

		res.print();
		env.execute();
	}
}
